package com.eshipenet.shipnet.edgechannelbuilder

import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.api.java.typeutils.TupleTypeInfo

class EdgeBoxKafkaDeserializer()
    extends KeyedDeserializationSchema[Tuple2[String, String]] {
  override def deserialize(messageKey: Array[Byte],
                           message: Array[Byte],
                           topic: String,
                           partition: Int,
                           offset: Long): Tuple2[String, String] =
    new Tuple2(new String(messageKey), new String(message))

  override def isEndOfStream(nextElement: Tuple2[String, String]): Boolean =
    false

  override def getProducedType: TypeInformation[Tuple2[String, String]] =
    new TupleTypeInfo[Tuple2[String, String]](BasicTypeInfo.STRING_TYPE_INFO,
                                              BasicTypeInfo.STRING_TYPE_INFO)
}
